package com.hitqz.robot.driver.zmq.protocol.component;

import lombok.extern.slf4j.Slf4j;
import org.zeromq.*;


/**
 * 处理ZeroMQ消息的线程
 */
@Slf4j
public class WorkerThread implements Runnable {

    private String backName;

    private String notePackageName;

    public WorkerThread(String backName, String notePackageName) {
        this.backName = backName;
        this.notePackageName = notePackageName;
    }


    @Override
    public void run() {
        try (ZContext ctx = new ZContext()) {
            ZMQ.Socket worker = ctx.createSocket(SocketType.DEALER);

            worker.setRcvHWM(0);
            worker.setSndHWM(0);

            worker.setTCPKeepAlive(1);
            worker.setTCPKeepAliveCount(3);
            worker.setTCPKeepAliveIdle(60L);
            worker.setTCPKeepAliveInterval(10L);

            worker.connect("ipc://" + backName);
            worker.send("ready".getBytes());

            ZMQObservableImpl zmqObservable = new ZMQObservableImpl();
            Class<?> className = Class.forName(notePackageName);
            zmqObservable.addObserver(className);
            while (!Thread.currentThread().isInterrupted()) {
                ZMsg msg = ZMsg.recvMsg(worker);
                ZFrame address = msg.pop();
                if (address.toString().startsWith("")) {
                    address = msg.pop();
                    if (address.toString().equals("ready")) {
                        address = msg.pop();
                    }
                }
                ZFrame type = msg.pop();
                if (!type.toString().equals("ready")) {
                    ZFrame data = msg.pop();
                    // 通知
                    zmqObservable.notifyObserver(address.toString(),type.toString(), data.getData());
                }
                // 重新唤醒
                worker.send("ready".getBytes());
            }
            log.info("over");
        } catch (Exception e) {
            log.info("[wheel][zmq]worker  发生错误:", e);
        }

    }
}
